package com.atguigu.gmall.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.utils.KafkaUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.StateTtlConfig.UpdateType;
import org.apache.flink.api.common.state.StateTtlConfig.StateVisibility;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.text.SimpleDateFormat;

/**
 * 访客 UV 计算
 * <p>
 * 需要启动的进程：zookeeper、kafka、hdfs、logger.sh、BaseLogApp、模拟生成日志的jar、UniqueVisitorApp
 * <p>
 * 程序执行流程：
 * <p>
 * 1、运行模拟生成日志的jar <br>
 * 2、将生成的日志 发送给 nginx ，nginx 进行负载均衡 <br>
 * 3、三台中的其中一台日志采集服务器接收到日志数据，将数据发送到 kafka topic：ods_base_log <br>
 * 4、BaseLogApp 从 topic：ods_base_log 读取数据，并使用侧输出流对其分流： <br>
 *    &nbsp&nbsp&nbsp&nbsp&nbsp&nbsp  ① 启动日志：dwd_start_log <br>
 *    &nbsp&nbsp&nbsp&nbsp&nbsp&nbsp  ② 曝光日志：dwd_display_log <br>
 *    &nbsp&nbsp&nbsp&nbsp&nbsp&nbsp  ③ 页面日志：dwd_page_log <br>
 * 5、UniqueVisitorApp 从 topic：dwd_page_log 读取数据，对 PV 数据进行过滤： <br>
 *    &nbsp&nbsp&nbsp&nbsp&nbsp&nbsp ① 按照 mid 进行分组 <br>
 *    &nbsp&nbsp&nbsp&nbsp&nbsp&nbsp ② 使用 filter 算子对数据进行过滤 <br>
 *    &nbsp&nbsp&nbsp&nbsp&nbsp&nbsp ③ 在过滤的时候，使用状态变量记录上次访问日期 <br>
 *
 * @author lvbingbing
 * @date 2022-03-26 17:01
 */
public class UniqueVisitorApp {
    public static void main(String[] args) throws Exception {
        // 1. 基本环境准备
        // 1.1 流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 1.2 并行度设置
        env.setParallelism(4);
        // 2. 检查点相关设置
        // 2.1 开启检查点
        env.enableCheckpointing(6000L, CheckpointingMode.EXACTLY_ONCE);
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        // 2.2 设置检查点超时时间
        checkpointConfig.setCheckpointTimeout(60000L);
        // 2.3 作业取消时保留检查点
        checkpointConfig.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 2.4 设置重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5000L));
        // 2.5 设置状态后端
        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/ck/gmall"));
        // 2.6 指定操作 HDFS 的用户
        System.setProperty("HADOOP_USER_NAME", "atguigu");
        // 3. 从 kafka 中读取数据
        String topic = "dwd_page_log";
        String groupId = "unique_visitor_app_group";
        FlinkKafkaConsumer<String> kafkaSource = KafkaUtils.getKafkaSource(topic, groupId);
        DataStreamSource<String> dataStreamSource = env.addSource(kafkaSource);
        // 4. 数据结构转换。String -> JSONObject
        SingleOutputStreamOperator<JSONObject> jsonObjectDs = dataStreamSource.map(JSON::parseObject);
        // 5. UV 统计核心逻辑
        // 5.1 根据 设备id 进行分组
        KeyedStream<JSONObject, String> keyedStream = jsonObjectDs.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));
        // 5.2 自定义 RichFilterFunction，对同一设备id在一天的范围内的访问页面请求进行过滤
        SingleOutputStreamOperator<JSONObject> uvDs = keyedStream.filter(new UvCountRichFilterFunction());
        // 6. 打印统计结果
        uvDs.print("uv 统计结果");
        // 7. 将过滤后的 UV 数据，写回到 kafka 的 dwm 层
        // todo 使用精准一次性
        uvDs.map(JSONAware::toJSONString)
                .addSink(KafkaUtils.getKafkaSinkAtLeastOnce("dwm_unique_visitor"));
        // 8. 触发程序执行
        env.execute();
    }

    /**
     * 对同一设备id在一天的范围内的访问页面请求进行过滤，去除重复的页面请求
     */
    public static class UvCountRichFilterFunction extends RichFilterFunction<JSONObject> {

        /**
         * 状态变量，用于存放上次访问日期
         */
        private transient ValueState<String> lastVisitedDateState;

        /**
         * 日期格式化对象
         */
        private SimpleDateFormat sdf;

        /**
         * 初始化状态与格式化对象
         *
         * @param configuration 配置信息
         * @throws Exception 异常信息
         */
        @Override
        public void open(Configuration configuration) throws Exception {
            // 1. 初始化日期格式化对象
            sdf = new SimpleDateFormat("yyyyMMdd");
            // 2. 初始化状态变量
            ValueStateDescriptor<String> lastVisitedStateDescriptor = new ValueStateDescriptor<>("lastVisitedDateState", String.class);
            // 配置状态存活时间为 1 天
            StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.days(1))
                    // 上次访问时间戳  在每次写入操作 创建和更新状态 时初始化(也是默认配置，可不设置)
                    .setUpdateType(UpdateType.OnCreateAndWrite)
                    // 状态过期后，即使还没有被 JVM 回收，也不返回给调用者(也是默认配置，可不设置)
                    .setStateVisibility(StateVisibility.NeverReturnExpired)
                    .build();
            lastVisitedStateDescriptor.enableTimeToLive(ttlConfig);
            lastVisitedDateState = getRuntimeContext().getState(lastVisitedStateDescriptor);
        }

        @Override
        public boolean filter(JSONObject jsonObject) throws Exception {
            // 1. 如果当前访问页面是从其他页面跳转过来的，返回 false，将其过滤掉
            String lastPageId = jsonObject.getJSONObject("page").getString("last_page_id");
            if (StringUtils.isNotEmpty(lastPageId)) {
                return false;
            }
            // 获取状态中的上次访问时间
            String lastVisitedDate = lastVisitedDateState.value();
            // 获取本次访问时间
            String curVisitDate = sdf.format(jsonObject.getLong("ts"));
            // 2. 如果上次已经访问过，并且上次访问时间与本次访问时间相等(同一天)，返回 false，将其过滤掉
            if (lastVisitedDate != null && lastVisitedDate.equals(curVisitDate)) {
                return false;
            }
            // 3. 第一次访问，更新状态，返回 true，保留该条记录
            lastVisitedDateState.update(curVisitDate);
            return true;
        }
    }
}
